package cn.com.nes.common.util;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.CancellationException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;

public class Promise<V> {

	public interface Inform<V> {
		void resolve(V v);
		void reject(Exception e);
	}
	
	private final AtomicReference<PromiseState> state = new AtomicReference<>(PromiseState.RUNNING);
	private final Queue<Thread> queue = new LinkedList<Thread>();
	private final Object mutex = new Object();
	
	private final Runnable cancel;
	
	private volatile V v;
	private volatile Exception e;
	private volatile Consumer<V> complete;
	private volatile Consumer<Exception> exceptionCaught;
	
	private enum PromiseState {
		CANCELLED,
		DONE,
		RUNNING
	}
	
	private final Inform<V> inform = new Inform<V>() {

		@Override
		public void resolve(V v) {
			if(Promise.this.state.compareAndSet(PromiseState.RUNNING, PromiseState.DONE)) {
				Promise.this.v = v;
				Promise.this.e = null;
				synchronized (mutex) {
					Thread th;
					while((th = queue.poll()) != null) {
						LockSupport.unpark(th);
					}
					if(Promise.this.complete != null && Promise.this.v != null) {
						Promise.this.complete.accept(v);
					}
					if(Promise.this.exceptionCaught != null && Promise.this.e != null) {
						Promise.this.exceptionCaught.accept(e);
					}
				}
			}
		}

		@Override
		public void reject(Exception e) {
			if(Promise.this.state.compareAndSet(PromiseState.RUNNING, PromiseState.DONE)) {
				Promise.this.v = null;
				Promise.this.e = e;
				synchronized (mutex) {
					Thread th;
					while((th = queue.poll()) != null) {
						LockSupport.unpark(th);
					}
					if(Promise.this.complete != null && Promise.this.v != null) {
						Promise.this.complete.accept(v);
					}
					if(Promise.this.exceptionCaught != null && Promise.this.e != null) {
						Promise.this.exceptionCaught.accept(e);
					}
				}
			}
		}
	};
	
	private Promise(Consumer<Inform<V>> consumer, Runnable cancel) {
		this.cancel = cancel;
		try {
			consumer.accept(inform);
		} catch(Exception e) {
			inform.reject(e);
		}
	}
	
	public static <V> Promise<V> newPromise(Consumer<Inform<V>> consumer) {
		return newPromise(consumer, null);
	}
	
	public static <V> Promise<V> newPromise(Consumer<Inform<V>> consumer, Runnable cancel) {
		return new Promise<V>(consumer, cancel);
	}

	public V get(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException, CancellationException, Exception {
		PromiseState state = this.state.get();
		if(state == PromiseState.CANCELLED) {
			throw new CancellationException();
		} else if(state == PromiseState.DONE) {
			if(this.e != null) {
				throw e;
			} else {
				return v;
			}
		} else {
			synchronized (mutex) {
				if(this.state.get() != PromiseState.RUNNING) {
					return this.get();
				}
				queue.offer(Thread.currentThread());
			}
			if(timeout > 0 && unit != null) {
				LockSupport.parkNanos(unit.toNanos(timeout));
			} else {
				LockSupport.park();  // 与unpark是成对的
			}
			synchronized(mutex) {
				queue.remove(Thread.currentThread());
			}
			if(Thread.interrupted()) {
				// 清除中断状态并抛出InterruptedException
				throw new InterruptedException();
			}
			state = this.state.get();
			if(state != PromiseState.RUNNING) {
				return this.get();
			} else {
//				cancel(false);  // 等待超时后取消这个任务
				throw new TimeoutException();
			}
		}
	}

	public V get() throws InterruptedException, Exception {
		return this.get(0, null);
	}

	public Promise<V> exceptionCaught(Consumer<Exception> call) {
		if(this.state.get() == PromiseState.DONE) {
			if(this.e != null) {
				call.accept(e);
			}
		} else {
			synchronized (mutex) {
				if(this.state.get() == PromiseState.DONE) {
					if(this.e != null) {
						call.accept(e);
					}
				} else {
					this.exceptionCaught = call;
				}
			}
		}
		return this;
	}

	public Promise<V> then(Consumer<V> call) {
		if(this.state.get() == PromiseState.DONE) {
			if(this.v != null) {
				call.accept(v);
			}
		} else {
			boolean flag = false;
			synchronized (mutex) {
				if(this.state.get() == PromiseState.DONE) {
					if(this.v != null) {
						flag = true;
					}
				} else {
					this.complete = call;
				}
			}
			if(flag) {
				call.accept(v);
			}
		}
		return this;
	}

	public void cancel(boolean mayInterruptIfRunning) {
		if(this.state.compareAndSet(PromiseState.RUNNING, PromiseState.CANCELLED)) {
			synchronized(mutex) {
				Thread th;
				while((th = queue.poll()) != null) {
					if(mayInterruptIfRunning) {
						th.interrupt();
					} else {
						LockSupport.unpark(th);
					}
				}
			}
			if(this.cancel != null) {
				cancel.run();
			}
		}
	}

	public boolean isCancelled() {
		PromiseState state = this.state.get();
		return state == PromiseState.CANCELLED;
	}

	public boolean isDone() {
		PromiseState state = this.state.get();
		return state == PromiseState.DONE;
	}
}
